//package com.main.bd.flink.sink;
//
//import org.apache.flink.api.common.serialization.SimpleStringEncoder;
//import org.apache.flink.connector.file.sink.FileSink;
//import org.apache.flink.core.fs.Path;
//import org.apache.flink.runtime.state.filesystem.FsStateBackend;
//import org.apache.flink.streaming.api.datastream.DataStreamSource;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
//import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
//import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
//
//import java.util.concurrent.TimeUnit;
//
//public class FileSink {
//
//    public static void main(String[] args) throws Exception {
//        //1.env
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.enableCheckpointing(TimeUnit.SECONDS.toMillis(10));
//        env.setStateBackend(new FsStateBackend("file:///D:/ckp"));
//
//        //2.source
//        DataStreamSource<String> lines = env.socketTextStream("node1", 9999);
//
//        //3.sink
//        //设置sink的前缀和后缀
//        //文件的头和文件扩展名
//        //prefix-xxx-.txt
//        OutputFileConfig config = OutputFileConfig
//                .builder()
//                .withPartPrefix("prefix")
//                .withPartSuffix(".txt")
//                .build();
//
//        //设置sink的路径
//        String outputPath = "hdfs://node1:8020/FlinkFileSink/parquet";
//
//        final FileSink<String> sink = FileSink
//                .forRowFormat(new Path(outputPath), new SimpleStringEncoder<String>("UTF-8"))
//                .withBucketAssigner(new DateTimeBucketAssigner<>())
//                .withRollingPolicy(
//                        DefaultRollingPolicy.builder()
//                                .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
//                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
//                                .withMaxPartSize(1024 * 1024 * 1024)
//                                .build())
//                .withOutputFileConfig(config)
//                .build();
//
//        lines.sinkTo(sink).setParallelism(1);
//
//        env.execute();
//    }
//
//}
